package com.study.mq.kafka.partioner;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.utils.Utils;

import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;

/**
 * @author ysy
 * @version 1.0
 * @Title: CustomPartitioner
 * @Note  自定义分区处理器
 * <br><b>PackageName:</b> com.study.mq.kafka.partioner
 * <br><b>ClassName:</b>
 * <br><b>Date:</b>
 */
public class MyPartitioner implements Partitioner {

    /**
     * @Author ysy
     * @Description  分区算法
     * @Date 17:40 [topic, key, keyBytes, value, valueBytes, cluster]
     * @Param
     * @return int
     **/
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        Integer partitionCount = cluster.partitionCountForTopic(topic);
        if(keyBytes == null){
            // 随机分区
            return Utils.toPositive(ThreadLocalRandom.current().nextInt()) % partitionCount;
        }else{
            // 保持和DefaultPartitioner一致，即murmur算法分区
            return Utils.toPositive(Utils.murmur2(keyBytes)) % partitionCount;
        }
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
